package com.stylefeng.guns.modular.orderpush.service.impl;

import com.stylefeng.guns.modular.orderpush.model.Order;
import com.stylefeng.guns.modular.orderpush.dao.OrderMapper;
import com.stylefeng.guns.modular.orderpush.service.IOrderService;
import com.baomidou.mybatisplus.service.impl.ServiceImpl;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author brunoli123
 * @since 2018-05-07
 */
@Service
public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements IOrderService {



    public void megProducer(){
        // 创建一个Produer Group
        DefaultMQProducer producer = new DefaultMQProducer("producerGroup");
        // 指定NameServer地址
        producer.setNamesrvAddr("118.89.236.58:9876");
        producer.setVipChannelEnabled(false);
        try {
            // 启动producer
            producer.start();
            // 创建一个Message ,并指定topic、Tag和消息主体
            Message msg = new Message("bruno", "TagA",
                    ("Hello RocketMQ ").getBytes(RemotingHelper.DEFAULT_CHARSET));
            //向broker发送一个消息
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        } catch (MQClientException | UnsupportedEncodingException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //当produer不再使用时,关闭produer
        producer.shutdown();
    }


    public void megConsumer()  {
        //指定一个Consumer Group 来创建一个consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");

        //指定NameServer 地址,consumer和NameServer建立长链接,并且获取topic信息以及broker的ip和地址
        consumer.setNamesrvAddr("118.89.236.58:9876");
        //指定消费offset的位置。TIMESTAMP表示从consumer建立后,producer向broker新发送的消息开始
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
        consumer.setVipChannelEnabled(false);
        //指定消费topic,和过滤的TAG
        try {
            consumer.subscribe("bruno", "*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        //注册一个回调函数,当comsumer接收到消息时,执行的动作
        List<MessageExt> msgList = new ArrayList<>();
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                //msgList = msgs;
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //启动comsumer实例
        try {
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        System.out.printf("Consumer Started.%n");
        //return ;
    }

}
